package org.rabbitmq.mq02.mqServer;

import jdk.jfr.DataAmount;
import lombok.Data;
import org.rabbitmq.mq02.common.Consumer;
import org.rabbitmq.mq02.common.MqException;
import org.rabbitmq.mq02.mqServer.core.*;
import org.rabbitmq.mq02.mqServer.datacore.DiskDataManager;
import org.rabbitmq.mq02.mqServer.datacore.MemoryDataCenter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 将内存和磁盘上的数据进行整合,用"虚拟机"这个概念将其整合起来.
 * 不同虚拟机中的交换机 队列,绑定关系,消息都是不互通的.
 * 此处为了简化,仅实现单台虚拟主机,但在数据结构上设置不同虚拟主句名
 * 为区分不同的虚拟主机上的设备,通过配置设备名区别:(以虚拟机名为前缀)
 * 规定:
 *  exchangeName = virtualHostName+exchangeName;
 *  queueName = virtualHostName+queueName;
 *  并且将调用的方法抛出的异常都在这个类中进行处理,不再向上抛出
 */
@Data
public class VirtualHost {
    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataManager diskDataManager = new DiskDataManager();

    public VirtualHost(String virtualHostName){
        this.virtualHostName = virtualHostName;
        //初始化磁盘数据:
        diskDataManager.init();
        //初始化内存数据
        try {
            memoryDataCenter.recovery(diskDataManager);
        } catch (IOException | MqException | ClassNotFoundException e) {
            System.out.println("[VirtualHost] 内存数据恢复失败");
            e.printStackTrace();
        }
    }

    //在对交换机在内存和磁盘上插入和删除数据时,可能存在线程安全问题,要以交换机为维度对其上锁
    //交换机锁对象:
    private final Object exchangeLocker = new Object();
    //交换机操作:
    //创建交换机,
    //创建后,将其保存到内存和磁盘上
    public boolean exchangeDeclare(String exchangeName, ExchangeType type, boolean durable,
                                   boolean autoDelete, Map<String,Object> args){
       //   先根据约定 设置交换机名
        exchangeName = virtualHostName + exchangeName;
        synchronized(exchangeLocker){
            //先在内存上查找,若已存在,则直接返回
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange!=null){
                System.out.println("[VirtualHost] 交换机已经存在,不再创建 exchangeName:"+exchangeName);
                return true;
            }
            exchange = new Exchange();
            exchange.setName(exchangeName);
            exchange.setType(type);
            exchange.setDurable(durable);
            exchange.setAutoDelete(autoDelete);
            //这里对args参数的设置.要在Exchange类中再为args关于Map参数添加set和get方法
            exchange.setArgs(args);
            //先存入数据库,再存入内存中,
            //这个顺序是:插入数据库操作比较容易出现异常,存内存出现异常的可能小较小
            //         若插入数据库失败,则不再存入内存中;
            //         若是转换顺序,当存数据库出现异常时,还要将内存中的数据再删了,比较麻烦
            if(durable){
                //当交换机设置为持久化时,将其存入内存:
                diskDataManager.insertExchange(exchange);
            }
            //存入内存
            memoryDataCenter.insertExchange(exchange);
            System.out.println("[VirtualHost] 交换机创建成功 exchangeName:"+exchangeName);
            return true;
        }
    }
    //删除交换机
    //在内存和磁盘上将数据删除
    public boolean exchangeDelete(String exchangeName){
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker){
                Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                if(exchange==null){
                    throw new MqException("[VirtualHost] 要删除的交换机不存在 exchangeName:"+exchangeName);
                }
                //删除内存数据:
                memoryDataCenter.deleteExchange(exchangeName);
                //删除磁盘数据:
                boolean durable = exchange.isDurable();
                if(durable){
                    diskDataManager.deleteExchange(exchangeName);
                }
                System.out.println("[VirtualHost] 交换机删除成功 exchangeName:"+exchangeName);
                return true;
            }
        } catch (MqException e) {
            System.out.println("[VirtualHost] 交换机删除失败 exchangeName:"+exchangeName);
            e.printStackTrace();
        }
        return false;
    }

    //在对队列在内存和磁盘上插入和删除数据时,可能存在线程安全问题,要以队列为维度对其上锁
    //创建 队列锁对象:
    private final Object queueLocker = new Object();
    /**队列
     * 创建队列:创建队列并将其存入到磁盘和内存中
     */
    public boolean queueDeclare(String queueName,boolean isDurable,boolean autoDelete, Map<String,Object> args){
        queueName = virtualHostName+queueName;
        try {
            synchronized(queueLocker){
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if(existsQueue!=null){
                    System.out.println("[VirtualHost] 队列已经存在 queueName:"+queueName);
                    return true;
                }
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(isDurable);
                queue.setAutoDelete(autoDelete);
                //此处在MSGQueue类中,针对args属性,要实现关于Map类型的set方法
                queue.setArgs(args);
                //存入磁盘
                if(isDurable) {
                    diskDataManager.insertQueue(queue);
                }
                //存入内存
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 创建队列成功 !");
            }
            return true;
        } catch (IOException | MqException e) {
            System.out.println("[VirtualHost] 创建队列失败 queueName:"+queueName);
            e.printStackTrace();
            return false;
        }
    }
    //删除队列:从磁盘和内存中 删除队列
    public boolean queueDelete(String queueName){
        queueName = virtualHostName+queueName;
        try{
            synchronized(queueLocker){
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if(existsQueue==null){
                    throw new MqException("[VirtualHost] 队列不存在,删除队列失败 queueName:"+queueName);
                }
                if(existsQueue.isDurable()){
                    diskDataManager.deleteQueue(queueName);
                }
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 删除队列成功 queueName:"+queueName);
            }
            return true;
        }catch (Exception e) {
            System.out.println("[VirtualHost] 队列删除失败! queueName:" +queueName);
            e.printStackTrace();
            return false;
        }
    }

//  该类实现和绑定相关的操作
    private Router router = new Router();
    //绑定的插入和删除
    //插入绑定
    public boolean bindingDeclare(String exchangeName,String queueName,String bindingKey){
        exchangeName = virtualHostName + exchangeName;
        queueName = virtualHostName + queueName;
        try{
            //1.验证绑定是否存在,不存在再创建
            Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
            if(binding!=null){
                throw new MqException("[VirtualHost] 绑定已存在 exchangeName:"+exchangeName+
                        " ,queueName:"+queueName);
            }
            //这里再创建一个类router,实现关于绑定相关的操作
            //2.判断bindingKey格式是否正确
            boolean ok = router.checkBindingKey(bindingKey);
            if(!ok){
                throw new MqException("[VirtualHost] 绑定格式有误 bindingKey:"+bindingKey);
            }
            //3.创建绑
            binding = new Binding();
            binding.setExchangeName(exchangeName);
            binding.setQueueName(queueName);
            binding.setBindingKey(bindingKey);
            //4.验证绑定的队列和交换机是否存在
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue==null){
                throw new MqException("[VirtualHost] 要绑定的队列不存在 queueName:"+queueName+
                        " ,bindingKey:"+bindingKey);
            }
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange==null){
                throw new MqException("[VirtualHost] 要绑定的交换机不存在 exchangeName:"+exchangeName+
                        " ,bindingKey:"+bindingKey);
            }
            //5,存入磁盘
            //当队列和交换机同时设置持久化时,将该绑定关系存入磁盘
            if(queue.isDurable() && exchange.isDurable()){
                diskDataManager.insertBinding(binding);
            }
            //6.存入内存
            memoryDataCenter.insertBinding(binding);
            System.out.println("[VirtualHost] 创建绑定成功 bindingKey: "+bindingKey);
            return true;
        }catch (MqException e) {
            System.out.println("[VirtualHost] 创建绑定失败 bindingKey:"+bindingKey);
            e.printStackTrace();
        }
        return false;
    }

    //删除绑定
    public boolean bindingDelete(String exchangeName,String queueName){
        exchangeName = virtualHostName + exchangeName;
        queueName = virtualHostName + queueName;
        try {
            Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
            if(binding==null){
                throw new MqException("[VirtualHost] 绑定不存在 queueName:"+queueName+
                        " ,exchangeName:"+exchangeName);
            }
            //从内存删除
            memoryDataCenter.deleteBinding(binding);
            //从磁盘删除
            //此处可能绑定没有保存在磁盘上,删除失败,但没有关系,没有影响
            diskDataManager.deleteBinding(binding.getBindingKey());
            System.out.println("[VirtualHost] 删除绑定成功 exchangeName:"+exchangeName+
                    " ,queueName:"+queueName +" , bindingKey:"+binding.getBindingKey());
            return true;
        }catch(Exception e){
            System.out.println("[VirtualHost] 删除绑定失败 exchangeName:"+exchangeName+
                    " ,queueName:"+queueName);
            e.printStackTrace();
            return false;
        }
    }

    //发送消息到队列
    public boolean basicPublish(String exchangeName,String routingKey,
                                BasicProperties basicProperties,byte[] body){
        exchangeName = virtualHostName + exchangeName;
        try {
            //1.判断交换机是否存在
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange==null){
                throw new MqException("[VirtualHost] 交换机不存在 exchangeName:"+exchangeName);
            }
            //2.判断routingKey格式是否正确
            boolean ok = router.checkRoutingKey(routingKey);
            if(!ok) {
                throw new MqException("[VirtualHost] routingKey格式有误 routingKey:"+routingKey);
            }
            //3.根据交换机的类型进行路由匹配,分发消息
            if(exchange.getType()==ExchangeType.DIRECT){
                //直接交换机,routingKey就是队列名,bindingKey无用,将消息路由到指定的队列上
                //获取到指定队列
                String queueName = virtualHostName + routingKey;
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue==null){
                    throw new MqException("[VirtualHost] 队列不存在 queueName:"+queueName);
                }
                //构造消息对象
                Message message = new Message();
                message = message.createMessageById(null,basicProperties,body);
                //发送消息到队列,再构造一个方法实现
                sendMessage(queue,message);
            }else{
                //当交换机类型为fanout/topic时:
                //遍历交换机所有的绑定
                ConcurrentHashMap<String, Binding> bindings = memoryDataCenter.getBindings(exchangeName);
                for(Binding b:bindings.values()){
                    MSGQueue queue = memoryDataCenter.getQueue(b.getQueueName());
                    //判断交换机绑定的队列是否存在:
                    if(queue==null){
                        System.out.println("[VirtualHost] 队列不存在 queueName:"+b.getQueueName());
                        continue;
                    }
                    //构造消息对象
                    Message message = new Message().createMessageById(routingKey, basicProperties, body);
                    //判断routingKey与binding是否成功
                    if(!router.isRouting(exchange.getType(),message.getRoutingKey(),b.getBindingKey())){
                       //匹配失败:
                        System.out.println("[VirtualHost] routingKey和BindingKey不匹配 routingKey:"+routingKey+
                                " , bindingKey:"+b.getBindingKey());
                        continue;
                    }
                    //匹配成功时,就将消息转发
                    sendMessage(queue,message);
                    System.out.println("[VirtualHost] 消息发送成功 queueName:"+queue.getName()+
                            " ,messageId:"+message.getMessageId());
                }
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost]消息发送失败 ");
            e.printStackTrace();
            return false;
        }
    }
    //消费者管理对象:
    private ConsumerManager consumerManager = new ConsumerManager(this);

    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
            //存入磁盘
            //是否持久化
            //1:持久化 0:非持久化
            if(message.getDeliveryMode()==1){
                diskDataManager.sendMessage(queue,message);
            }
            //存入内存:
            memoryDataCenter.sendMessage(queue,message);
            //消息已经到达队列,通知订阅队列的消费者消费消息
            consumerManager.notifyConsumer(queue.getName());
            System.out.println("[VirtualHost] 发送消息成功");
    }


    /**
    //订阅消息
    //添加一个订阅者:
     * @param consumerTag 消费者身份标识
     * @param queueName 队列名
     * @param autoAck 是否自动确认消息
     * @param consumer 回调函数
     * @return
     */
    public boolean basicConsume(String consumerTag, String queueName,
                                 boolean autoAck, Consumer consumer){
        queueName = virtualHostName + queueName;
        try {
            //通过消费者管理类实现添加消费者功能
            consumerManager.addConsumer(consumerTag,queueName,autoAck,consumer);
            System.out.println("[VirtualHost] basicConsumer 成功 queueName:"+queueName);
            return true;
        }  catch (MqException e) {
            System.out.println("[VirtualHost] basicConsumer 失败 queueName:"+queueName);
            e.printStackTrace();
            return false;
        }
    }

    //手动确认消费消息时,调用该方法
    public boolean basicAck(String queueName,String messageId){
        try {
            //1.获取消息
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue==null){
                throw new MqException("[VirtualHost] 队列不存在 queueName:"+queueName);
            }
            //2.获取队列
            Message message = memoryDataCenter.getMessage(messageId);
            if(message==null){
                throw new MqException("[VirtualHost] 消息不存在 messageId:"+messageId);
            }
            //删除磁盘数据
            if(message.getDeliveryMode()==1){
                diskDataManager.deleteMessageFromQueue(queue,message);
            }
            //删除未确认消息队列集合中的消息
            memoryDataCenter.deleteWaitMessage(queueName,messageId);
            //删除消息集合中的数据
            memoryDataCenter.deleteMessage(messageId);
            System.out.println("[VirtualHost] basicAck 消息被成功确认 queueName:"+queueName
                    +" ,messageId:"+messageId);
            return true;
        } catch (MqException | IOException | ClassNotFoundException e) {
            System.out.println("[VirtualHost] basicAck 消息确认失败 queueName:"+queueName
                    +" ,messageId:"+messageId);
            e.printStackTrace();
            return false;
        }
    }



}
